/*
 * ServerDaemon.cpp
 *
 *  Created on: Mar 18, 2012
 *      Author: Victor Nidens victornlj@gmail.com
 * 		Author:	Mikhail S. Ananyevskiy, PhD, msaipme@yandex.ru
 *
 *	This file is part of Cloud Mechatronic Laboratory (CMLab) software.
 *
 *	Cloud Mechatronic Laboratory (CMLab) for CMLab is free software: you can redistribute it and/or modify
 *	it under the terms of the GNU Lesser General Public License as published by
 *	the Free Software Foundation, either version 3 of the License, or
 *	(at your option) any later version.
 *
 *	Cloud Mechatronic Laboratory (CMLab) software is distributed in the hope that it will be useful,
 *	but WITHOUT ANY WARRANTY; without even the implied warranty of
 *	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *	GNU Lesser General Public License for more details.
 *
 *	You should have received a copy of the GNU Lesser General Public License
 *	along with Cloud Mechatronic Laboratory (CMLab). If not, see <http://www.gnu.org/licenses/>.
 */

#include "ServerDaemon.h"
#include <iostream>
#include <string.h>
#include <signal.h>
#include <cstdlib>
#include <sys/wait.h>
#include <arpa/inet.h>
#include <algorithm>
#include <fcntl.h>
#include <string>

ServerDaemon::ServerDaemon(ServerConfig conf){
	memset(&config, 0, sizeof(ServerConfig));
	memcpy(&config, &conf, sizeof(ServerConfig));

	memset(&serverUdpSocket, 0, sizeof(serverUdpSocket));

	serverUdpSocket.sin_family = AF_INET;
	serverUdpSocket.sin_addr.s_addr = htonl(INADDR_ANY);
	serverUdpSocket.sin_port = htons(config.udpPort);

	serverUdpSocketFD = socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, 0);

	memset(&serverTcpSocket, 0, sizeof(serverTcpSocket));

	serverTcpSocket.sin_family = AF_INET;
	serverTcpSocket.sin_addr.s_addr = htonl(INADDR_ANY);
	serverTcpSocket.sin_port = htons(config.tcpPort);

	serverTcpSocketFD = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);

	memset(&serverRegulatorSocket, 0, sizeof(serverRegulatorSocket));

	serverRegulatorSocket.sin_family = AF_INET;
	serverRegulatorSocket.sin_addr.s_addr = htonl(INADDR_ANY);
	serverRegulatorSocket.sin_port = htons(config.regulatorPort);

	serverRegulatorSocketFD = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);

	initDBStorage();

}

int ServerDaemon::bindSockets(){

	int bindResult;

	bindResult = bind(serverUdpSocketFD, (struct sockaddr*) &serverUdpSocket, sizeof(serverUdpSocket));

	if (bindResult == -1) {
		cerr << "Can't bind clients UDP socket" << endl;
		return -1;
	}

	bindResult = bind(serverTcpSocketFD, (struct sockaddr*) &serverTcpSocket, sizeof(serverTcpSocket));

	if (bindResult == -1) {
		cerr << "Can't bind clients TCP socket" << endl;
		return -1;
	}

	bindResult = listen(serverTcpSocketFD, 50);

	if (bindResult == -1) {
		cerr << "Can't listen clients TCP socket" << endl;
		return -1;
	}

	bindResult = bind(serverRegulatorSocketFD, (struct sockaddr*) &serverRegulatorSocket, sizeof(serverRegulatorSocket));

	if (bindResult == -1) {
		cerr << "Can't bind regulators TCP socket" << endl;
		return -1;
	}

	bindResult = listen(serverRegulatorSocketFD, 50);

	if (bindResult == -1) {
		cerr << "Can't listen regulators TCP socket" << endl;
		return -1;
	}

	return 0;

}

int ServerDaemon::initDBStorage(){

	if(config.savePacketStat){

		dbConnection = mysql_init(NULL);

		if(!mysql_real_connect(dbConnection,
								config.dbHost.c_str(),
								config.dbUser.c_str(),
								config.dbUserPassword.c_str(),
								config.dbName.c_str(),
								config.dbPort,
								NULL, 0)){
			cerr << "Can't connect to MySQL db" << endl;
			config.savePacketStat = false;
			return 1;
		}
	}

	return 0;
}

void ServerDaemon::savePacketStat(ExchangePacket *packet){

	//if statistics save flag is switched off then just return
	if(!config.savePacketStat)
		return;

	char buf[16] = {0};

	string insertQuery = "INSERT INTO " + config.dbTable +
						" (sendTimestampSec, sendTimestampUsec, receiveTimestampSec, receiveTimestampUsec, sourceName, sourceIP, commandId, commandName, nxtName)" +
						" values (";

	//send timestamp seconds
	sprintf(buf, "%llu", packet->source.timestampSec);
	insertQuery.append(buf);
	insertQuery.append(", ");

	//send timestamp microseconds
	memset(buf, 0, sizeof(buf));
	sprintf(buf, "%llu", packet->source.timestampUsec);
	insertQuery.append(buf);
	insertQuery.append(", ");

	//received timestamp seconds
	memset(buf, 0, sizeof(buf));
	sprintf(buf, "%lu", packet->receivedTimestamp.tv_sec);
	insertQuery.append(buf);
	insertQuery.append(", ");

	//received timestamp microseconds
	memset(buf, 0, sizeof(buf));
	sprintf(buf, "%lu", packet->receivedTimestamp.tv_usec);
	insertQuery.append(buf);
	insertQuery.append(", ");
	insertQuery.append("'");

	//source name
	insertQuery.append(packet->source.sourceName);
	insertQuery.append("', '");

	//ip
	sprintf(buf, "%i.%i.%i.%i", packet->source.sourceIp[0], packet->source.sourceIp[1], packet->source.sourceIp[2], packet->source.sourceIp[3]);
	insertQuery.append(buf);
	insertQuery.append("', ");

	//command id
	memset(buf, 0, sizeof(buf));
	sprintf(buf, "%i", packet->command.command);
	insertQuery.append(buf);
	insertQuery.append(", '");

	//command name
	insertQuery.append(commandName(packet->command.command));
	insertQuery.append("', '");

	//nxt name
	insertQuery.append(packet->deviceData.deviceName);
	insertQuery.append("')");

	int query_state = mysql_query(dbConnection, insertQuery.c_str());

	if (query_state != 0) {
		cerr << mysql_error(dbConnection) << endl;
		return;
	}

}

void ServerDaemon::exec(){

	if(bindSockets()) {
		cerr << "Server daemon not started." << endl;
		return;
	}

	daemonize();
}

void ServerDaemon::daemonize(){
//	int pid;
//	struct sigaction sa;

//	cout << "::: Starting Server daemon :::\n" << flush;
//	pid = fork();

//	switch (pid) {
//	case 0:
//		setsid();
//		chdir("/");
//
//		close(0);
//		close(1);
//		close(2);
//
//		memset(&sa, 0, sizeof(sa));
//		sa.sa_handler = &sighandler;
//		sigaction(SIGCHLD, &sa, 0);

//		openlog("daemon2", 0, LOG_USER);
		mainloop();
//		closelog();

		exit(0);

//	case -1:
//		cout << "[ERR]Couldn't start server daemon" << endl;
//		break;
//
//	default:
//		cout << "OK. PID=" << pid << endl;
//		break;
//	}
}

void ServerDaemon::sighandler(int signum) {
    waitpid(0, 0, WNOHANG);
}

void ServerDaemon::mainloop(){

	fd_set readset;
	timeval timeout;
	int maxFD;
	uint8 buffer[128];
	struct sockaddr_in udpSource;
	unsigned int udpAddrLen = sizeof(struct sockaddr_in);
	ExchangePacket receivedPacket;
	ExchangePacket sentPacket;
	string name;

	timeout.tv_sec = 0;
	timeout.tv_usec = 1000;

	while(1){

		FD_ZERO(&readset);

		FD_SET(serverTcpSocketFD, &readset);
		FD_SET(serverUdpSocketFD, &readset);
		FD_SET(serverRegulatorSocketFD, &readset);

		for(HashSet<int>::iterator it = openedTcpSocketsFD.begin(); it != openedTcpSocketsFD.end(); it++){
			FD_SET(*it, &readset);
		}

		for(HashSet<int>::iterator it = openedRegulatorsSocketsFD.begin(); it != openedRegulatorsSocketsFD.end(); it++){
			FD_SET(*it, &readset);
		}

		maxFD = max(serverTcpSocketFD, max(serverUdpSocketFD, serverRegulatorSocketFD));	//recalculate maxFD properly

		if(openedTcpSocketsFD.size() > 0){
			maxFD = max(maxFD, *max_element(openedTcpSocketsFD.begin(), openedTcpSocketsFD.end()));
		}

		if(openedRegulatorsSocketsFD.size() > 0){
			maxFD = max(maxFD, *max_element(openedRegulatorsSocketsFD.begin(), openedRegulatorsSocketsFD.end()));
		}

		if(select(maxFD + 1, &readset, NULL, NULL, &timeout) < 0){
			//TODO: write to error log
			break;
		}

		//new client connection initiated
		if(FD_ISSET(serverTcpSocketFD, &readset)){
			int sock = accept(serverTcpSocketFD, NULL, NULL);

			if(sock < 0){
				//TODO: write to error log
				cerr << "Can't accept new CLIENT connection" << endl;
				return;
			}

			fcntl(sock, F_SETFL, O_NONBLOCK);

//			cout << "[TCP CONN] New client connected" << endl;

			openedTcpSocketsFD.insert(sock);
		}

		//new regulator connection initiated
		if(FD_ISSET(serverRegulatorSocketFD, &readset)){
			int sock = accept(serverRegulatorSocketFD, NULL, NULL);

			if(sock < 0){
				//TODO: write to error log
				cerr << "Can't accept new REGULATOR connection" << endl;
				return;
			}

			fcntl(sock, F_SETFL, O_NONBLOCK);

//			cout << "[SERVER TCP CONN] New regulator connected" << endl;

			openedRegulatorsSocketsFD.insert(sock);
		}

		//check if there is something new in the UDP socket
		if(FD_ISSET(serverUdpSocketFD, &readset)){
			int received = recvfrom(serverUdpSocketFD, buffer, sizeof(buffer), 0, (struct sockaddr*)&udpSource, (socklen_t*)&udpAddrLen);

//			cout << "[SERVER UDP INCOME][" << received << " bytes] Something received on UDP socket" << endl;

			if(received == NETWORK_PACKET_SIZE){

//				cout << "[SERVER UDP PACK INCOME] Received ExchangePacket on UDP socket" << endl;

				memset(&receivedPacket, 0, sizeof(receivedPacket));

				memcpy(&receivedPacket, buffer, NETWORK_PACKET_SIZE);

				gettimeofday(&receivedPacket.receivedTimestamp, NULL);

				//save name and address of the packet source
				connectedUdpClients[receivedPacket.source.sourceName] = udpSource;

				clientsPacketsQueue.push_back(receivedPacket);

				savePacketStat(&receivedPacket);

			}
		}

		//check if there is something new in some client TCP socket
		for(HashSet<int>::iterator it = openedTcpSocketsFD.begin(); it != openedTcpSocketsFD.end(); it++){
			int val = *it;
			if(FD_ISSET(val, &readset)){

				int received = read(*it, buffer, NETWORK_PACKET_SIZE);

				if(received <= 0){
					close(*it);
					openedTcpSocketsFD.erase(*it);
					continue;
				}

//				cout << "[SERVER TCP INCOME][" << received << " bytes] Something received on TCP socket" << endl;

				if(received == NETWORK_PACKET_SIZE){
//					cout << "[SERVER TCP PACK INCOME] Received ExchangePacket on TCP socket" << endl;

					memset(&receivedPacket, 0, sizeof(receivedPacket));

					memcpy(&receivedPacket, buffer, NETWORK_PACKET_SIZE);

					gettimeofday(&receivedPacket.receivedTimestamp, NULL);

					//save name and address of the packet source
					connectedTcpClients[receivedPacket.source.sourceName] = *it;

					clientsPacketsQueue.push_back(receivedPacket);

					savePacketStat(&receivedPacket);

				}

			}
		}

		//check if there is something new in some regulator TCP socket
		for(HashSet<int>::iterator it = openedRegulatorsSocketsFD.begin(); it != openedRegulatorsSocketsFD.end(); it++){
			if(FD_ISSET(*it, &readset)){
				int received = read(*it, buffer, NETWORK_PACKET_SIZE);

				if(received <= 0){
					close(*it);
					openedRegulatorsSocketsFD.erase(it);
				}

//				cout << "[SERVER REGULATOR INCOME][" << received << " bytes] Something received on regulator TCP socket" << endl;

				if(received == NETWORK_PACKET_SIZE){
//					cout << "[SERVER REGULATOR PACK INCOME] Received ExchagePacket on regulator TCP socket" << endl;

					memset(&receivedPacket, 0, sizeof(receivedPacket));

					memcpy(&receivedPacket, buffer, NETWORK_PACKET_SIZE);

					gettimeofday(&receivedPacket.receivedTimestamp, NULL);

					//save name and address of the packet source
					connectedRegulators[receivedPacket.source.sourceName] = *it;

					regulatorsRequestQueue.push_back(receivedPacket);

					savePacketStat(&receivedPacket);

				}
			}
		}

		handleRegulatorsRequests();
		handleClientsRequests();

	}

}

void ServerDaemon::handleRegulatorsRequests(){

	if(regulatorsRequestQueue.size() < 1){
		return;
	}

	//fetch one request from regulator queue and remove it from queue
	vector<ExchangePacket>::iterator regulatorIt = regulatorsRequestQueue.begin();

	ExchangePacket regulatorPacket = *regulatorIt;

	ExchangePacket sendPacket;

	uint8 buffer[128] = {0};
	int res = 0;

	regulatorsRequestQueue.erase(regulatorIt);

	if(regulatorPacket.command.command == COMMAND_PING)
		return;

	if(regulatorPacket.command.command == COMMAND_READ_MOTOR_REQUEST
			|| regulatorPacket.command.command == COMMAND_READ_SENSOR_REQUEST){
		awaitingClientResponse.push_back(regulatorPacket);
	}

//	cout << "Size of awaitingClientResponse is [" << awaitingClientResponse.size() << "]\n";

	memset(&sendPacket, 0, sizeof(sendPacket));

	fillSourceInfo(&sendPacket.source, config.ip, config.name.c_str(), regulatorPacket.command.responseProtocol);		//construct new source info

	memcpy(&sendPacket.command, &regulatorPacket.command, sizeof(Command));					//copy command data

	memcpy(&sendPacket.deviceData, &regulatorPacket.deviceData, sizeof(Device));			//copy device control data

	memcpy(buffer, &sendPacket, NETWORK_PACKET_SIZE);										//map packet to send buffer

	if(sendPacket.source.sourceProtocol == PROTOCOL_TCP || sendPacket.source.sourceProtocol == PROTOCOL_SAME){
		//send request to the corresponding client through TCP socket

		if(connectedTcpClients.find(sendPacket.command.dst) == connectedTcpClients.end())
			return;

		int clientFD = connectedTcpClients[sendPacket.command.dst];

		if(clientFD <= 0)
			return;

		res = send(clientFD, buffer, NETWORK_PACKET_SIZE, 0);

		if(res != NETWORK_PACKET_SIZE){
			//TODO: notify an error occurred
		}

	} else if(sendPacket.source.sourceProtocol == PROTOCOL_UDP) {
		//send request to the corresponding client through UDP socket

		if(connectedUdpClients.find(sendPacket.command.dst) == connectedUdpClients.end())
			return;

		struct sockaddr_in clientAddr = connectedUdpClients[sendPacket.command.dst];

		res = sendto(serverUdpSocketFD, buffer, NETWORK_PACKET_SIZE, 0, (sockaddr*)&clientAddr, sizeof(clientAddr));

		if(res != NETWORK_PACKET_SIZE){
			//TODO: notify an error occurred
		}

	}

}

void ServerDaemon::handleClientsRequests(){

	if(clientsPacketsQueue.size() < 1)
		return;

	//fetch one client request from the queue

	vector<ExchangePacket>::iterator clientIt = clientsPacketsQueue.begin();
	ExchangePacket clientPacket = *clientIt;
	ExchangePacket sendPacket;
	int regulatorFD = 0;
	int res;
	uint8 buffer[128] = {0};

	clientsPacketsQueue.erase(clientIt);

	//TODO: save packet info to DB

	//if it's just a ping packet then send ping response
	if(clientPacket.command.command == COMMAND_PING){
		fillSourceInfo(&sendPacket.source,
						config.ip,
						config.name.c_str(),
						clientPacket.command.responseProtocol == PROTOCOL_SAME ? clientPacket.source.sourceProtocol : clientPacket.command.responseProtocol);

		constructCommand(&sendPacket.command, COMMAND_PING, clientPacket.source.sourceName, 0, 0, PROTOCOL_SAME);

		memcpy(buffer, &sendPacket, NETWORK_PACKET_SIZE);

		if(sendPacket.source.sourceProtocol == PROTOCOL_TCP){
			int clientFD = 0;

			if(connectedTcpClients.find(sendPacket.command.dst) == connectedTcpClients.end())
				return;

			clientFD = connectedTcpClients[sendPacket.command.dst];

			if(clientFD <= 0)
				return;

			res = send(clientFD, buffer, NETWORK_PACKET_SIZE, 0);

			if(res != NETWORK_PACKET_SIZE){
				//TODO: notify an error occurred
			}

//		} else if(sendPacket.source.sourceProtocol == PROTOCOL_UDP){

		} else {
			if(connectedUdpClients.find(sendPacket.command.dst) == connectedUdpClients.end())
				return;

			struct sockaddr_in clientAddr = connectedUdpClients[sendPacket.command.dst];

			res = sendto(serverUdpSocketFD, buffer, NETWORK_PACKET_SIZE, 0, (sockaddr*)&clientAddr, sizeof(clientAddr));

			if(res != NETWORK_PACKET_SIZE){
				//TODO: notify an error occurred
			}

		}

		return;

	}

	if(clientPacket.command.command == COMMAND_DEVICE_ANNOUNCE){

		availableNXTDevices[clientPacket.deviceData.deviceName] = clientPacket.source.sourceName;

		memset(&sendPacket, 0, sizeof(sendPacket));

		fillSourceInfo(&sendPacket.source, config.ip, config.name, clientPacket.command.responseProtocol);

		constructCommand(&sendPacket.command, COMMAND_DEVICE_ANNOUNCE_RESPONSE, clientPacket.source.sourceName, 0, 0, PROTOCOL_SAME);

		memcpy(sendPacket.deviceData.deviceName, clientPacket.deviceData.deviceName, sizeof(sendPacket.deviceData.deviceName));

		memcpy(buffer, &sendPacket, NETWORK_PACKET_SIZE);

		if(sendPacket.source.sourceProtocol == PROTOCOL_TCP){
			int clientFD = 0;

			if(connectedTcpClients.find(sendPacket.command.dst) == connectedTcpClients.end())
				return;

			clientFD = connectedTcpClients[sendPacket.command.dst];

			if(clientFD <= 0)
				return;

			res = send(clientFD, buffer, NETWORK_PACKET_SIZE, 0);

			if(res != NETWORK_PACKET_SIZE){
				//TODO: notify an error occurred
			}

		} else {
			if (connectedUdpClients.find(sendPacket.command.dst) == connectedUdpClients.end())
				return;

			struct sockaddr_in clientAddr =	connectedUdpClients[sendPacket.command.dst];

			res = sendto(serverUdpSocketFD, buffer, NETWORK_PACKET_SIZE, 0,	(sockaddr*) &clientAddr, sizeof(clientAddr));

			if (res != NETWORK_PACKET_SIZE) {
				//TODO: notify an error occurred
			}
		}

		return;
	}

//	cout << "Size of awaitingClientResponse is [" << awaitingClientResponse.size() << "]";
	cout << endl;

	//find corresponding regulator request
	for(vector<ExchangePacket>::iterator it = awaitingClientResponse.begin(); it != awaitingClientResponse.end(); it++){
		if(it->command.command == (clientPacket.command.command - 1)
				&& !memcmp(it->command.dst, clientPacket.source.sourceName, 16)
				&& !memcmp(it->deviceData.deviceName, clientPacket.deviceData.deviceName, 16)){
			regulatorFD = connectedRegulators[it->source.sourceName];
			awaitingClientResponse.erase(it);
			break;
		}
	}

	if(regulatorFD <= 0)
		return;

	memset(&sendPacket, 0, sizeof(sendPacket));

	fillSourceInfo(&sendPacket.source, config.ip, config.name.c_str(), PROTOCOL_TCP);

	memcpy(&sendPacket.command, &clientPacket.command, sizeof(Command));

	memcpy(&sendPacket.deviceData, &clientPacket.deviceData, sizeof(Device));

	memcpy(buffer, &sendPacket, NETWORK_PACKET_SIZE);

	res = send(regulatorFD, buffer, NETWORK_PACKET_SIZE, 0);

	if(res != NETWORK_PACKET_SIZE){
		//TODO: notify an error occurred
	}

}
